package com.ebspp.client.placement;

import java.io.File;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;

import com.ebspp.client.serverManager.AsyncGBCollector;
import com.ebspp.global.EBSPPTimer;
import com.ebspp.global.General;
import com.ebspp.thrift.DataMsg;
import com.ebspp.thrift.Identity;
import com.ebspp.thrift.ResourceType;
import com.sleepycat.je.Database;
import com.sleepycat.je.DatabaseConfig;
import com.sleepycat.je.DatabaseEntry;
import com.sleepycat.je.Environment;
import com.sleepycat.je.EnvironmentConfig;
import com.sleepycat.je.LockMode;
import com.sleepycat.je.Transaction;

/**
 * It uses two separate db, one transactional lostDB for recording the servers->timestamp information.
 * placement mapping of keys written to the server after that timestamp is not persistent.
 * Hence, the recovery process should ask for the keys written after that timestamp from the corresponding
 * servers by using recover(ID, timestamp, server)
 * 
 * Another db is placementDB. it is not transactional, hence, data might get lost.
 * However, it is synced periodically. The client knows where to ask for the unsynced mappings
 * by reading the lostDB.
 * 
 * note that key and timestamp are concantenated as key + ":)" + timestamp, and stored in bdb as a key
 * @author x23pan
 *
 */
public class MvDBPlacement implements IPlacement
{
	Identity id;
	ResourceType resType;
	
	HashSet<String> lostServers;
	
	Environment placementEnv;
	Database placementDB;
	String placementDir;
	
	Environment lostEnv;
	Database lostDB;
	String lostDir;
	
	public MvDBPlacement(Identity id, ResourceType type) throws Exception
	{
		this.id = id;
		this.resType = type;
		
		//gbCollector = new AsyncGBCollector(id, type);
		lostServers = new HashSet<String>();
		
		//don't change the create order
		createPlacementDB();
		createLostDB();
	}
	
	private void createPlacementDB() throws Exception
	{
		placementDir = General.dir + "/placement/placement/";
		File pFile = new File(placementDir);
		if(!pFile.exists())
		{
			if(pFile.mkdirs())
				System.out.println("Create placement dir at:" + placementDir);
			else
				throw new Exception("Unable to create placement dir at: " + placementDir);
		}

		EnvironmentConfig pEnvConf = new EnvironmentConfig();
		pEnvConf.setAllowCreate(true);
		placementEnv = new Environment(pFile, pEnvConf);
		
		DatabaseConfig pDBConf = new DatabaseConfig();
		pDBConf.setAllowCreate(true);
		placementDB = placementEnv.openDatabase(null, "placement", pDBConf);
	}
	
	private void createLostDB() throws Exception
	{
		lostDir= General.dir + "/placement/lost/";
		File lFile = new File(lostDir);
		if(!lFile.exists())
		{
			if(lFile.mkdirs())
				System.out.println("Create lost dir at:" + lostDir);
			else
				throw new Exception("Unable to create lost dir at:" + lostDir);
		}
		else
		{
			//todo
			//if(lostDB.count() > 0)
				//findLostMapping();
		}
		
		EnvironmentConfig lEnvConf = new EnvironmentConfig();
		lEnvConf.setAllowCreate(true);
		lEnvConf.setTransactional(true);
		lostEnv = new Environment(lFile, lEnvConf);
		
		DatabaseConfig lDBConf = new DatabaseConfig();
		lDBConf.setAllowCreate(true);
		lDBConf.setTransactional(true);
		lostDB = lostEnv.openDatabase(null, "lost", lDBConf);
	}
	
	private void writeLostDB(String server)
	{
		Transaction txn = lostEnv.beginTransaction(null, null);
		lostDB.put(txn, new DatabaseEntry(server.getBytes()), 
							new DatabaseEntry(General.getTS().getBytes()));
		txn.commit();
	}
	
	/**
	 * set the placement mapping of key->server
	 * it is essential if you don't want to lose your data
	 */
	@Override
	public void setPlace(String server, List<DataMsg> data) 
	{
		// TODO Auto-generated method stub
		if(!lostServers.contains(server))
		{
			writeLostDB(server);
			lostServers.add(server);
		}
		
		for(DataMsg msg : data)
		{
			String dbKey = msg.getKey();
			if(msg.getTs() != null)
				dbKey += ":)" + msg.getTs();
			placementDB.put(null, new DatabaseEntry(dbKey.getBytes()), new DatabaseEntry(server.getBytes()));
		}
	}


	@Override
	public String getPlace(DataMsg msg) 
	{
		// TODO Auto-generated method stub
		String dbKey = msg.getKey();
		if(msg.getTs() != null)
			dbKey += ":)" + msg.getTs();
		DatabaseEntry keyEntry = new DatabaseEntry(dbKey.getBytes());
		DatabaseEntry valEntry = new DatabaseEntry();
		
		placementDB.get(null, keyEntry, valEntry, LockMode.DEFAULT);
		
		byte[] server = valEntry.getData();
		if(server == null)
			return null;
		
		return new String(server);
	}


	/**
	 * The checkpointer sync the placementDB periodically.
	 * After the each checkpoint, mappings before are safe and persistent
	 * @author x23pan
	 *
	 */
	class MvDBPCheckpointer implements Runnable
	{
		MvDBPlacement dbp;
		public MvDBPCheckpointer(MvDBPlacement dbp)
		{
			this.dbp = dbp;
		}
		
		@Override
		public void run() 
		{
			// TODO Auto-generated method stub
			while(true)
			{
				try {
					Thread.sleep(EBSPPTimer.checkpointPlacementT);
				} catch (InterruptedException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
				//copy all the lost servers. allow new lost servers added during sync	
				HashSet<String> servers = new HashSet<String>();
				synchronized(lostServers)
				{
					servers.addAll(lostServers);
					lostServers.clear();
				}
				
				/**
				 * if new lost server write txn happens before the delete txn,
				 * it's entry will be deleted. But it's data is safely synced.  
				 */
				Transaction txn = lostEnv.beginTransaction(null, null);
				for(String server : servers)
					lostDB.delete(txn, new DatabaseEntry(server.getBytes()));
				placementDB.sync();
				txn.commit();
			}
		}
	}

}
